package org.eleorc.projectreactor;

import org.eleorc.flow.FlowDemo02;
import org.reactivestreams.Subscription;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;

import java.io.IOException;
import java.time.Duration;

public class FluxDemo {

    public static class MySubscriber<T> extends BaseSubscriber<T> {
        @Override
        protected void hookOnSubscribe(Subscription subscription) {
            System.out.println("MySubscriber hookOnSubscribe");
            request(1);
        }

        @Override
        protected void hookOnNext(T value) {
            System.out.println(Thread.currentThread().getName() + ":" + value);
            request(1);
        }
    }

    public static void main(String[] args) throws IOException {

//        test01();

//        test02();

//        test03();

        // 背压管理
        Flux.range(1,1000)
                .delayElements(Duration.ofMillis(10))
                .onBackpressureBuffer(1)
                .delayElements(Duration.ofMillis(9))
                .subscribe(new MySubscriber<Integer>());

        System.in.read();


    }

    private static void test03() throws IOException {
        //        Flux.just(1, 2, 3, 4, 5).buffer(2).subscribe(new MySubscriber<>());
//        Flux.range(1,20).limitRequest(3)
//        .subscribe(new MySubscriber<>());

        Flux.range(1,20).limitRate(3)
                .subscribeOn(Schedulers.boundedElastic())
                .subscribe(new MySubscriber<Integer>());

        System.in.read();
    }

    private static void test02() throws IOException {
        Flux<Integer> flux = Flux.just(1, 2, 3, 4, 5);

        BaseSubscriber<Integer> subscriber = new BaseSubscriber<Integer>() {
            @Override
            protected void hookOnSubscribe(Subscription subscription) {
                System.out.println(Thread.currentThread().getName() + ":" + this + " hookOnSubscribe");
                request(3);
//                super.hookOnSubscribe(subscription);
            }

            @Override
            protected void hookOnNext(Integer value) {
                System.out.println(Thread.currentThread().getName() + ":" + this + " hookOnNext =>" + value);
            }

            @Override
            protected void hookOnComplete() {
                super.hookOnComplete();
            }

            @Override
            protected void hookOnError(Throwable throwable) {
                super.hookOnError(throwable);
            }
        };

        flux.subscribe(subscriber);

        System.in.read();
    }

    /**
     * main:1
     * main:2
     * main:3
     * main:4
     * main:5
     * 并没有开启新线程
     */
    private static void test01() {
        Flux<Integer> flux = Flux.just(1, 2, 3, 4, 5);
        flux.subscribe((item) -> {
            if (item == 3) {
                try {
                    Thread.sleep(3000);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
            System.out.println(Thread.currentThread().getName() + ":" + item);
        });
    }
}
